package com.flink.datastreamapi.aggregation;

import com.flink.datastreamapi.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByDemo {

    public static void main(String[] args) throws Exception {

        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为2
        env.setParallelism(2);
        //数据源
        DataStreamSource<WaterSensor> dataStreamSource = env.fromElements(
                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_2", 21L, 21),
                new WaterSensor("sensor_2", 22L, 22),
                new WaterSensor("sensor_3", 3L, 3));

        //数据处理--分组（分区）
        // 要点：
        // 1.返回的是一个KeyedStream键控流
        //2.keyby不是转换算子，只能对数据进行重分区，不能设置并行度
        //3.keyby 分组和分区的关系
        //   3.1 keyby 是对数据分组，保证相同key的数据在同一个分区
        //   3.2 分区： 一个子任务，可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）
        //方式一： 使用匿名类来实现KeySelector
        KeyedStream<WaterSensor, String> keyedStream1 = dataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                //安装id作为key，并按照这个字段来分组
                return value.id;
            }
        });


        //方式二 使用Lambda表达式
        //  KeyedStream<WaterSensor, String> keyedStream2 = dataStreamSource.keyBy(value -> value.id);

        keyedStream1.print();
        //执行
        env.execute();


    }
}
